package org.example.reactor;

import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.util.function.Consumer;

public class BackpressureExample {
    public static void main(String[] args) throws InterruptedException {
        Flux.range(1, 1000)
                .doOnNext(i -> System.out.println("Producing " + i))
                .publishOn(Schedulers.newSingle("myThread"))
                .subscribe(new SlowSubscriber());

        Thread.sleep(5000);
    }

    static class SlowSubscriber implements Consumer<Integer> {
        private Subscription subscription;
        private int count = 0;

        @Override
        public void accept(Integer i) {
            sleep(100);
            System.out.println("Consuming " + i);
            count++;
            if (count % 10 == 0) {
                subscription.request(10);
            }
        }

        public void onSubscribe(Subscription subscription) {
            this.subscription = subscription;
            subscription.request(1);
        }

        private void sleep(int millis) {
            try {
                Thread.sleep(millis);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}